package org.apache.spark.examples.streaming;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.regex.Pattern;

/**
 * Created by admin on 2019/3/23.
 * $ ./bin/run-example streaming.JavaNetworkWordCount localhost 9999
 */
public class JavaNetworkWordCount {
    private static final Pattern SPACE = Pattern.compile(" ");
    public static void main(String[] args) throws Exception {
        SparkConf conf =new SparkConf()
                .setMaster("local[n]")
                .setAppName("NetworkWordCount");

        JavaStreamingContext jssc = new JavaStreamingContext(conf,
                Durations.seconds(1));

        //使用此上下文，我们可以创建一个DStream来表示来自TCP源的流数据，指定为主机名（例如localhost）和端口（例如9999）。
        JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost",9999);
        //此linesDStream表示将从数据服务器接收的数据流。此流中的每条记录都是一行文本。然后，我们想要将空格分割为单词。
        /*
        flatMap是一个DStream操作，它通过从源DStream中的每个记录生成多个新记录来创建新的DStream。

         */
        JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());

        /**
         * 使用PairFunction 对象将wordsDStream进一步映射（一对一转换）到(word, 1)对的DStream。
         * 然后，使用Function2对象将其缩小以获得每批数据中的单词频率。最后，wordCounts.print()将打印每秒生成的一些计数。
         */

        JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));
        JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey((i1, i2) -> i1 + i2);
        wordCounts.print();
        jssc.start();            //开始计算
        jssc.awaitTermination(); //等待计算终止

    }
}
